package org.pentaho.streaming.autotask;

import org.pentaho.di.core.row.RowMeta;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.StepDataInterface;
import org.pentaho.di.trans.step.StepInterface;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.step.StepMetaInterface;
import org.pentaho.di.trans.streaming.common.BaseStreamStep;
import org.pentaho.di.trans.streaming.common.FixedTimeStreamWindow;

public class AutoTaskConsumer extends BaseStreamStep implements StepInterface {
    private static final Class<?> PKG = AutoTaskConsumer.class; // for i18n purposes

    public AutoTaskConsumer(StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta,
                            Trans trans) {
        super(stepMeta, stepDataInterface, copyNr, transMeta, trans);
    }

    public boolean init(StepMetaInterface stepMetaInterface, StepDataInterface stepDataInterface) {
        boolean init = super.init(stepMetaInterface, stepDataInterface);
        AutoTaskConsumerMeta AutoTaskConsumerMeta = (AutoTaskConsumerMeta) this.variablizedStepMeta;

        try {
            RowMeta rowMeta = AutoTaskConsumerMeta.getRowMeta(getStepname(), this);
            window = new FixedTimeStreamWindow<>(subtransExecutor, rowMeta, getDuration(), getBatchSize());
            source = new AutoTaskStreamSource(AutoTaskConsumerMeta, this);
        } catch (Exception e) {
            getLogChannel().logError("AutoTaskConsumer Error", e);
            init = false;
        }
        return init;
    }
}
